Cookbook

Examples and Integration Recipes

Ready-to-run snippets from examples/ and integration patterns for sync/async, retries, and persistent sessions.

Examples in This Repository

File	Main Method	Purpose
`examples/run_simple.py`	`client.run()`	Basic prompt execution and final message output.
`examples/run_json_events.py`	`json_output=True`	Collect events, thread id, turn status, usage.
`examples/run_live_stream.py`	`client.run_live()`	Stream JSON events from a running sync process.
`examples/run_thread_session.py`	`start_thread()`	Create session and continue with follow-up prompts.
`examples/run_with_schema.py`	`run_with_schema()`	Constrain output to JSON schema and parse final JSON.
`examples/run_async.py`	`run_async()` + `run_live_async()`	Async run plus async event streaming.
`examples/run_persistent_session_store.py`	`JsonFileSessionStore`	Persist named sessions and resume by name.

Quick Recipes

1. Plain Prompt (Minimal)

from codex_local_sdk import CodexLocalClient

client = CodexLocalClient()
result = client.run_prompt("Summarize this project in 5 bullets.")
print(result.final_message)

2. Explicit Request with Sandbox

from codex_local_sdk import CodexExecRequest, CodexLocalClient, SandboxMode

client = CodexLocalClient()
result = client.run(
    CodexExecRequest(
        prompt="Find architecture risks.",
        sandbox=SandboxMode.READ_ONLY,
        skip_git_repo_check=True,
    ),
    timeout_seconds=120,
)

Async Pattern

import asyncio
from codex_local_sdk import CodexExecRequest, CodexLocalClient


async def main() -> None:
    client = CodexLocalClient()

    result = await client.run_async(CodexExecRequest(prompt="Analyze repo"), timeout_seconds=120)
    print(result.final_message)

    live = await client.run_live_async(CodexExecRequest(prompt="Stream repo analysis", json_output=True))
    async for event in live.iter_events():
        print(event.type)
    final = await live.wait()
    print(final.turn_status)


asyncio.run(main())

Retry/Backoff Pattern

from codex_local_sdk import CodexExecRequest, CodexLocalClient, RetryPolicy

client = CodexLocalClient(
    retry_policy=RetryPolicy(
        max_attempts=3,
        initial_backoff_seconds=0.5,
        backoff_multiplier=2.0,
        max_backoff_seconds=4.0,
        retry_on_exit_codes=(7, 9),
        jitter_ratio=0.2,
        max_total_retry_seconds=15.0,
        retry_on_timeouts=True,
    )
)

result = client.run(CodexExecRequest(prompt="Run resiliently"), timeout_seconds=30)
print(result.return_code, result.ok)

Persistent Session Pattern

from codex_local_sdk import CodexLocalClient, JsonFileSessionStore, SandboxMode

client = CodexLocalClient(session_store=JsonFileSessionStore(".codex_sessions.json"))

session, first = client.start_thread(
    prompt="Create migration plan",
    sandbox=SandboxMode.READ_ONLY,
    session_name="migration-plan",
)
print(session.session_id, first.turn_status)

second = client.resume(
    prompt="Continue with phase 1 tasks",
    session_name="migration-plan",
    last=False,
    json_output=True,
    timeout_seconds=120,
)
print(second.final_message)

same_session = client.open_session("migration-plan")
third = same_session.continue_prompt("Continue with test strategy", json_output=True)
print(third.turn_status)

Telemetry Hook Pattern

from codex_local_sdk import CodexClientEvent, CodexExecRequest, CodexLocalClient


def on_event(event: CodexClientEvent) -> None:
    print(event.type, event.operation, event.attempt, event.return_code)


client = CodexLocalClient(event_hook=on_event)
client.run(CodexExecRequest(prompt="Analyze repo"), timeout_seconds=60)

Service Integration Skeleton

class CodexService:
    def __init__(self) -> None:
        self.client = CodexLocalClient(default_cwd="/workspace/project")

    def analyze_repo(self) -> str:
        result = self.client.run_prompt(
            "Provide architecture summary and top 5 technical risks.",
            json_output=True,
        )
        return result.final_message or ""

    async def analyze_repo_async(self) -> str:
        result = await self.client.run_async(
            CodexExecRequest(prompt="Provide async architecture summary", json_output=True)
        )
        return result.final_message or ""